package com.yeziji.config;

import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.extra.spring.SpringUtil;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.rabbitmq.client.Channel;
import com.yeziji.common.business.dataMessage.base.PlatformUniqueInfo;
import com.yeziji.common.business.system.SystemConfigConvertor;
import com.yeziji.common.business.system.constant.enums.SystemConfigTypeEnum;
import com.yeziji.common.business.system.entity.SystemConfigEntity;
import com.yeziji.common.business.system.service.SystemConfigService;
import com.yeziji.common.rabbitMq.RabbitMqTopicEnum;
import com.yeziji.common.rabbitMq.RabbitMqTopicHandler;
import com.yeziji.constant.RabbitMqExchangeTypeEnum;
import com.yeziji.utils.expansion.Opt2;
import lombok.Builder;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.boot.context.event.ApplicationStartedEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.Order;

import javax.annotation.Nonnull;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.TimeoutException;

/**
 * 环境信息保持器
 * <p>用于保留各项转换信息</p>
 *
 * @author hwy
 * @since 2024/10/27 0:32
 **/
@Slf4j
@Order
@Configuration
public class EnvHolder implements ApplicationListener<ApplicationStartedEvent> {
    @Resource
    private ConnectionFactory connectionFactory;
    @Resource
    private SystemConfigService systemConfigService;
    @Resource
    private SimpleMessageListenerContainer simpleMessageListenerContainer;
    private static String PLATFORM_CODE = "";
    private static final String DEAD_LETTER_TAG = "dead.letter";
    private static final Cache<String, Object> COMBINE_CACHE_HOLDER =
            CacheBuilder.newBuilder()
                    // 动态调整初始化大小
                    .initialCapacity(2)
                    .maximumSize(30)
                    // 内存开始不足时开始回收
                    .softValues()
                    // 允许并发缓存 16 个线程
                    .concurrencyLevel(16)
                    .build();
    private static final Map<RabbitMqTopicEnum, RabbitMqTopicHandler> RABBIT_MQ_TOPIC_HOLDER = new HashMap<>();

    @Override
    public void onApplicationEvent(@Nonnull ApplicationStartedEvent event) {
        this.initRabbitMqHolder();
    }

    // --- init

    /**
     * 初始化 rabbit mq 保持器
     *
     * @see #RABBIT_MQ_TOPIC_HOLDER
     */
    private void initRabbitMqHolder() {
        // init holder handler
        Map<String, RabbitMqTopicHandler> beansOfType = SpringUtil.getApplicationContext().getBeansOfType(RabbitMqTopicHandler.class);
        for (Map.Entry<String, RabbitMqTopicHandler> rabbitEntry : beansOfType.entrySet()) {
            RabbitMqTopicHandler handler = rabbitEntry.getValue();
            RabbitMqTopicEnum topicEnum = rabbitEntry.getValue().getTopicEnum();
            RABBIT_MQ_TOPIC_HOLDER.put(topicEnum, handler);
        }

        // refresh platform code
        PLATFORM_CODE = "";
        SystemConfigEntity platformConfig = systemConfigService.getByTypeEnum(SystemConfigTypeEnum.PLATFORM_CODE);
        if (platformConfig != null) {
            PlatformUniqueInfo platformUniqueInfo = SystemConfigConvertor.init(platformConfig.getConfigValue()).buildConfig(PlatformUniqueInfo.class);
            PLATFORM_CODE = platformUniqueInfo.getLocalCode();
        }

        // init rabbit mq queues
        List<String> queueNames = new ArrayList<>();
        Set<RabbitMqTopicEnum> rabbitMqTopicEnums = rabbitMqTopics();
        for (RabbitMqTopicEnum rabbitMqTopicEnum : rabbitMqTopicEnums) {
            // 相同交换器的状态下, 根据平台生成不同的 routeKey 以及 queue
            String exchange = rabbitMqTopicEnum.getExchange();
            String routeKey = getAsRabbitPlatformKey(rabbitMqTopicEnum.getRouteKey());
            String queue = getAsRabbitPlatformKey(rabbitMqTopicEnum.getQueue());
            RabbitMqExchangeTypeEnum exchangeType = rabbitMqTopicEnum.getExchangeType();
            try (Connection connection = connectionFactory.createConnection();
                 Channel targetChannel = connection.createChannel(false)) {
                // 默认参数
                Map<String, Object> args = new LinkedHashMap<>();
                if (MapUtil.isNotEmpty(rabbitMqTopicEnum.getArgs())) {
                    args.putAll(rabbitMqTopicEnum.getArgs());
                }

                // 为每个客户端的主题，建立其独立的死信队列(允许 uniqueCode 为空)
                DeadLetterChannel deadLetterChannel = createDeadLetterChannel(PLATFORM_CODE, targetChannel, rabbitMqTopicEnum);
                if (deadLetterChannel != null) {
                    String queueName = deadLetterChannel.getQueueName();
                    if (!queueNames.contains(queueName)) {
                        queueNames.add(queueName);
                    }
                    args.putAll(deadLetterChannel.getArgs());
                }

                // 创建消息通道
                targetChannel.exchangeDeclare(exchange, exchangeType.getValue(), true);
                targetChannel.queueDeclare(queue, true, false, false, args);
                targetChannel.queueBind(queue, exchange, routeKey);
                if (!queueNames.contains(queue)) {
                    queueNames.add(queue);
                }
            } catch (IOException e) {
                log.error("[init rabbit mq] create channel exception: {}", e.getMessage(), e);
            } catch (TimeoutException e) {
                log.error("[init rabbit mq] create channel timeout exception: {}", e.getMessage(), e);
            }
        }

        // 设置消息队列
        simpleMessageListenerContainer.setQueueNames(queueNames.toArray(String[]::new));
    }

    // --- getter

    /**
     * 根据主题获取指定的处理器
     *
     * @param rabbitMqTopicEnum rabbit mq 主题
     * @return {@link RabbitMqTopicHandler}
     */
    public static RabbitMqTopicHandler getAsRabbitMqHandler(RabbitMqTopicEnum rabbitMqTopicEnum) {
        return RABBIT_MQ_TOPIC_HOLDER.get(rabbitMqTopicEnum);
    }

    /**
     * 根据交换机以及路由获取对应的处理器
     *
     * @param exchange 交换器名称
     * @param routeKey 路由名称
     * @return {@link RabbitMqTopicHandler}
     */
    public static RabbitMqTopicHandler getAsRabbitMqHandler(String exchange, String routeKey) {
        if (StrUtil.isBlank(exchange) || StrUtil.isBlank(routeKey)) {
            return null;
        }
        // cache key
        String key = String.format("rabbitMqHolder$%s_%s", exchange, routeKey);
        // 从缓存获取
        Object holderHandler = COMBINE_CACHE_HOLDER.getIfPresent(key);
        if (holderHandler == null) {
            final String finalExchange = exchange;
            final String finalRouteKey = routeKey;
            holderHandler = RABBIT_MQ_TOPIC_HOLDER.entrySet().stream()
                    .filter(entry -> {
                                RabbitMqTopicEnum topicEnum = entry.getKey();
                                String topicExchange = topicEnum.getExchange();
                                String topicRouteKey = getAsRabbitPlatformKey(topicEnum.getRouteKey());
                                // 获取当前客户端的死信队列标识
                                String deadExchange = String.format("%s.%s", DEAD_LETTER_TAG, topicExchange);
                                String deadRouteKey = getAsRabbitPlatformKey(String.format("%s.%s", DEAD_LETTER_TAG, topicEnum.getRouteKey()));
                                return (Objects.equals(topicExchange, finalExchange) || Objects.equals(deadExchange, finalExchange)) &&
                                        (Objects.equals(topicRouteKey, finalRouteKey) || Objects.equals(deadRouteKey, finalRouteKey));
                            }
                    )
                    .findFirst()
                    .map(Map.Entry::getValue)
                    .orElse(null);
        }

        // set cache
        if (holderHandler != null) {
            COMBINE_CACHE_HOLDER.put(key, holderHandler);
            RabbitMqTopicHandler rabbitMqTopicHandler = (RabbitMqTopicHandler) holderHandler;
            // holder handler 是同一个对象，所以每次需要独立判断 dead 标志
            rabbitMqTopicHandler.setDead(exchange.startsWith(DEAD_LETTER_TAG));
            return rabbitMqTopicHandler;
        }
        return null;
    }

    /**
     * 获取 rabbit mq 的所有主题枚举
     *
     * @return {@link Set}
     */
    public static Set<RabbitMqTopicEnum> rabbitMqTopics() {
        return RABBIT_MQ_TOPIC_HOLDER.keySet();
    }


    /**
     * 创建客户端的死信渠道
     *
     * @return {@link Map}
     */
    public static DeadLetterChannel createDeadLetterChannel(String uniqueCode,
                                                            Channel targetChannel,
                                                            RabbitMqTopicEnum rabbitMqTopicEnum) {
        Map<String, Object> argsMap = new HashMap<>();
        String exchange = rabbitMqTopicEnum.getExchange();
        String routeKey = rabbitMqTopicEnum.getRouteKey();
        String queue = rabbitMqTopicEnum.getQueue();
        if (StrUtil.isBlank(exchange) || StrUtil.isBlank(routeKey)) {
            return null;
        }

        // 为每个主题创建独立的死信队列
        uniqueCode = Opt2.nullEmpty(uniqueCode);
        boolean hasCode = StrUtil.isNotBlank(uniqueCode);
        String deadExchange = String.format("%s.%s", DEAD_LETTER_TAG, exchange);
        String deadRouteKey = hasCode ? String.format("%s.%s.%s", uniqueCode, DEAD_LETTER_TAG, routeKey) : String.format("%s.%s", DEAD_LETTER_TAG, routeKey);
        String deadQueue = hasCode ? String.format("%s.%s.%s", uniqueCode, DEAD_LETTER_TAG, queue) : String.format("%s.%s", DEAD_LETTER_TAG, queue);
        // 绑定到原主题上
        argsMap.put("x-dead-letter-exchange", deadExchange);
        argsMap.put("x-dead-letter-routing-key", deadRouteKey);
        try {
            // 创建死信队列
            targetChannel.exchangeDeclare(deadExchange, RabbitMqExchangeTypeEnum.DIRECT.getValue(), true);
            targetChannel.queueDeclare(deadQueue, true, false, false, null);
            targetChannel.queueBind(deadQueue, deadExchange, deadRouteKey);
            return DeadLetterChannel.builder()
                    .queueName(deadQueue)
                    .args(argsMap)
                    .build();
        } catch (IOException e) {
            log.error("[{}] 创建死信队列失败!!", rabbitMqTopicEnum);
        }
        return null;
    }

    private static String getAsRabbitPlatformKey(String key) {
        boolean hasCode = StrUtil.isNotBlank(PLATFORM_CODE);
        if (hasCode) {
            return String.format("%s.%s", PLATFORM_CODE, key);
        }
        return key;
    }

    @Data
    @Builder
    public static class DeadLetterChannel {
        private String queueName;
        private Map<String, Object> args;
    }
}